/**
 * Copyright 2019 Inc.
 **/
package com.myz.rabbitmq.common.retry;


import com.myz.rabbitmq.common.constants.Constants;
import com.myz.rabbitmq.common.dto.DetailResponse;
import com.myz.rabbitmq.common.dto.MessageWithTime;
import com.myz.rabbitmq.common.producer.MessageSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 重试机制
 *
 * @author maoyz0621 on 19-4-14
 * @version: v1.0
 */
public class RetryCache {

    private static final Logger logger = LoggerFactory.getLogger(RetryCache.class);

    private MessageSender messageSender;

    private boolean stop = false;

    /**
     * ConcurrentSkipListMap键值对按照key的顺序来存储
     */
    private Map<String, MessageWithTime> map = new ConcurrentSkipListMap<>();

    private AtomicLong id = new AtomicLong(0);

    public void setMessageSender(MessageSender messageSender) {
        this.messageSender = messageSender;
        startRetry();
    }

    /**
     * id 生成
     */
    public Long generateId() {
        return id.incrementAndGet();
    }

    /**
     * 添加消息
     *
     * @param messageWithTime
     */
    public void add(MessageWithTime messageWithTime) {
        map.putIfAbsent(messageWithTime.getId().toString(), messageWithTime);
    }

    /**
     * 移除消息
     *
     * @param id
     */
    public void del(String id) {
        map.remove(id);
    }


    /**
     * 尝试重试
     */
    private void startRetry() {
        new Thread(() -> {
            while (!stop) {
                try {
                    Thread.sleep(Constants.RETRY_TIME_INTERVAL);
                } catch (InterruptedException e) {
                    logger.error("{}", e);
                }
            }

            long now = System.currentTimeMillis();

            // 遍历MessageWithTime
            for (Map.Entry<String, MessageWithTime> entry : map.entrySet()) {
                MessageWithTime message = entry.getValue();
                if (message != null) {
                    if (message.getTime() + 3 * Constants.VALID_TIME < now) {  // 超时三分钟
                        logger.error("send message {} failed after 3 min ", message);
                        del(entry.getKey());
                    } else if (message.getTime() + Constants.VALID_TIME < now) { // 超时一分钟
                        DetailResponse response = messageSender.send(message);

                        // 若没有发送成功
                        if (!response.isSuccess()) {
                            logger.error("retry send message failed {} errMsg {}", message, response.getErrorMessage());
                        }
                    }
                }
            }

        }).start();
    }
}
